package com.yanggu.mapreduce.mapjoin;

import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.*;
import java.net.URI;
import java.util.HashMap;
import java.util.Map;

/**
 * @version V1.0
 * @author: YangGu
 * @date: 2020/2/8 14:01
 * @description:
 */
public class MapJoinMapper extends Mapper<LongWritable, Text, Text, NullWritable> {

    private Map<String, String> map = new HashMap<>();

    private Text k = new Text();

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        //1. 将小表中的数据缓存到内存中
        URI[] cacheFiles = context.getCacheFiles();
        String path = cacheFiles[0].getPath();
        BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(new FileInputStream(new File(path))));
        String line;
        while ((line = bufferedReader.readLine()) != null) {
            String[] fields = line.split("\t");
            map.put(fields[0], fields[1]);
        }

        //2. 关闭资源
        IOUtils.closeStream(bufferedReader);
    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //1. 得到一行数据
        String line = value.toString();

        //2. 切割数据
        String[] fields = line.split("\t");

        //3. 封装数据
        String kString = fields[0];

        //这里可以处理left join的问题。如果map中没有对应的pid，使用null字符串代替
        if (map.containsKey(fields[1])) {
            kString = kString.concat("\t" + map.get(fields[1]));
        } else {
            kString = kString.concat("\t" + "null");
        }

        kString = kString.concat("\t" + fields[2]);

        k.set(kString);

        context.write(k, NullWritable.get());
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        //这里执行一些关闭资源的操作。例如销毁map和map中的数据
        map.clear();
        map = null;
    }
}
